* @param string $db DB name
* @param string $id ID string of transaction
* @param float $writeTime Time spent in write queries
+ * @param integer $affected Number of rows affected by writes
*/
- public function transactionWritingOut( $server, $db, $id, $writeTime = 0.0 ) {
+ public function transactionWritingOut( $server, $db, $id, $writeTime = 0.0, $affected = 0 ) {
$name = "{$server} ({$db}) (TRX#$id)";
if ( !isset( $this->dbTrxMethodTimes[$name] ) ) {
$this->logger->info( "Detected no transaction for '$name' - out of sync." );
);
$slow = true;
}
+ // Warn if too many rows were changed...
+ if ( $affected > $this->expect['maxAffected'] ) {
+ $this->reportExpectationViolated(
+ 'maxAffected',
+ "[transaction $id writes to {$server} ({$db})]",
+ $affected
+ );
+ }
// Fill in the last non-query period...
$lastQuery = end( $this->dbTrxMethodTimes[$name] );
if ( $lastQuery ) {
* @var integer Number of write queries for the current transaction
*/
private $mTrxWriteQueryCount = 0;
+ /**
+ * @var integer Number of rows affected by write queries for the current transaction
+ */
+ private $mTrxWriteAffectedRows = 0;
/**
* @var float Like mTrxWriteQueryCount but excludes lock-bound, easy to replicate, queries
*/
return $this->mTrxLevel ? $this->mTrxWriteCallers : [];
}
+ public function pendingWriteRowsAffected() {
+ return $this->mTrxWriteAffectedRows;
+ }
+
/**
* Get the list of method names that have pending write queries or callbacks
* for this transaction
if ( $ret !== false ) {
$this->lastPing = $startTime;
if ( $isWrite && $this->mTrxLevel ) {
- $this->updateTrxWriteQueryTime( $sql, $queryRuntime );
+ $this->updateTrxWriteQueryTime( $sql, $queryRuntime, $this->affectedRows() );
$this->mTrxWriteCallers[] = $fname;
}
}
*
* @param string $sql A SQL write query
* @param float $runtime Total runtime, including RTT
+ * @param integer $affected Affected row count
*/
- private function updateTrxWriteQueryTime( $sql, $runtime ) {
+ private function updateTrxWriteQueryTime( $sql, $runtime, $affected ) {
// Whether this is indicative of replica DB runtime (except for RBR or ws_repl)
$indicativeOfReplicaRuntime = true;
if ( $runtime > self::SLOW_WRITE_SEC ) {
$this->mTrxWriteDuration += $runtime;
$this->mTrxWriteQueryCount += 1;
+ $this->mTrxWriteAffectedRows += $affected;
if ( $indicativeOfReplicaRuntime ) {
$this->mTrxWriteAdjDuration += $runtime;
$this->mTrxWriteAdjQueryCount += 1;
$this->mTrxShortId = sprintf( '%06x', mt_rand( 0, 0xffffff ) );
$this->mTrxWriteDuration = 0.0;
$this->mTrxWriteQueryCount = 0;
+ $this->mTrxWriteAffectedRows = 0;
$this->mTrxWriteAdjDuration = 0.0;
$this->mTrxWriteAdjQueryCount = 0;
$this->mTrxWriteCallers = [];
if ( $this->mTrxDoneWrites ) {
$this->mLastWriteTime = microtime( true );
$this->trxProfiler->transactionWritingOut(
- $this->mServer, $this->mDBname, $this->mTrxShortId, $writeTime );
+ $this->mServer,
+ $this->mDBname,
+ $this->mTrxShortId,
+ $writeTime,
+ $this->mTrxWriteAffectedRows
+ );
}
$this->runOnTransactionIdleCallbacks( self::TRIGGER_COMMIT );
$this->mTrxAtomicLevels = [];
if ( $this->mTrxDoneWrites ) {
$this->trxProfiler->transactionWritingOut(
- $this->mServer, $this->mDBname, $this->mTrxShortId );
+ $this->mServer,
+ $this->mDBname,
+ $this->mTrxShortId
+ );
}
$this->mTrxIdleCallbacks = []; // clear